package io.sunny.platform.ai.deepseek.processor;

import java.time.Duration;
import java.util.List;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * @Author SUNNY
 * @Date 2025/11/30 12:53
 * @PackageName: io.sunny.platform.ai.deepseek.processor
 * @Description: 批量处理器，用于将多个请求合并后统一处理。
 *               @param <T> 请求类型
 *               @param <R> 响应类型
 * @Version 1.0
 */
public class BatchProcessor<T, R> {
    private final int batchSize;
    private final int bufferSize;
    private final Duration maxWaitTime;
    private final BatchProcessor.Processor<T, R> processor;
    private final BlockingQueue<Batch<T>> batchQueue;
    private final Thread processingThread;
    private final AtomicBoolean running = new AtomicBoolean(true);
    private final AtomicInteger processedCount = new AtomicInteger(0);

    // 内部类：表示一个批次的数据
    private static class Batch<T> {
        final List<T> items;
        final long timestamp;

        Batch(List<T> items) {
            this.items = items;
            this.timestamp = System.currentTimeMillis();
        }
    }

    // 处理器接口
    public interface Processor<T, R> {
        List<R> process(List<T> items);
    }

    // 构造函数私有化，使用 Builder 创建实例
    private BatchProcessor(
            int batchSize,
            int bufferSize,
            Duration maxWaitTime,
            Processor<T, R> processor) {
        this.batchSize = batchSize;
        this.bufferSize = bufferSize;
        this.maxWaitTime = maxWaitTime;
        this.processor = processor;
        this.batchQueue = new LinkedBlockingQueue<>(bufferSize);
        this.processingThread = new Thread(this::processBatches);
        this.processingThread.start();
    }

    // 启动批处理线程
    private void processBatches() {
        while (running.get()) {
            try {
                // 等待直到有批次或超时
                Batch<T> batch = batchQueue.poll(maxWaitTime.toMillis(), TimeUnit.MILLISECONDS);
                if (batch != null) {
                    // 处理批次
                    List<R> results = processor.process(batch.items);
                    // 可以在这里添加结果处理逻辑
                    processedCount.addAndGet(batch.items.size());
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                break;
            }
        }
    }

    // 提交单个请求
    public void submit(T item) {
        if (item == null) return;

        // 尝试放入缓冲区
        if (batchQueue.offer(new Batch<>(List.of(item)))) {
            // 触发检查是否需要处理
            checkAndProcess();
        } else {
            // 缓冲区满，直接处理
            List<R> results = processor.process(List.of(item));
            processedCount.incrementAndGet();
        }
    }

    // 提交多个请求
    public void submitAll(List<T> items) {
        if (items == null || items.isEmpty()) return;

        // 如果数量超过 batch size，则分批提交
        for (int i = 0; i < items.size(); i += batchSize) {
            int end = Math.min(i + batchSize, items.size());
            List<T> subList = items.subList(i, end);
            if (batchQueue.offer(new Batch<>(subList))) {
                checkAndProcess();
            } else {
                // 缓冲区满，直接处理
                List<R> results = processor.process(subList);
                processedCount.addAndGet(subList.size());
            }
        }
    }

    // 检查是否需要处理批次
    private void checkAndProcess() {
        // 实际实现中可以优化为更高效的机制
        // 这里简化处理，仅作为示意
    }

    // 获取已处理的数量
    public int getProcessedCount() {
        return processedCount.get();
    }

    // 关闭处理器
    public void close() {
        running.set(false);
        try {
            processingThread.join();
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }

    // Builder 类
    public static class Builder<T, R> {
        private int batchSize = 10;
        private int bufferSize = 100;
        private Duration maxWaitTime = Duration.ofMillis(50);
        private Processor<T, R> processor;

        public Builder<T, R> batchSize(int batchSize) {
            this.batchSize = batchSize;
            return this;
        }

        public Builder<T, R> bufferSize(int bufferSize) {
            this.bufferSize = bufferSize;
            return this;
        }

        public Builder<T, R> maxWaitTime(Duration maxWaitTime) {
            this.maxWaitTime = maxWaitTime;
            return this;
        }

        public Builder<T, R> processor(Processor<T, R> processor) {
            this.processor = processor;
            return this;
        }

        public BatchProcessor<T, R> build() {
            if (processor == null) {
                throw new IllegalStateException("Processor must be set");
            }
            return new BatchProcessor<>(batchSize, bufferSize, maxWaitTime, processor);
        }
    }
}
